package nr.nodehost.service;

import java.io.*;
import java.net.*;
import java.util.ArrayList;
import java.util.Map;
import java.util.Vector;
import java.util.concurrent.ConcurrentHashMap;

import lombok.extern.slf4j.Slf4j;
import nr.nodehost.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;

import net.sf.json.JSONObject;
import nr.nodehost.model.*;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Service
@Component
@EnableAsync
@Slf4j
public class AsyncTaskService {
    public static final int BUF_LENGTH = 65535;
    private static ConcurrentHashMap<String, Socket> componentId_sockets = new ConcurrentHashMap<>();   //sockets映射集合
    //Add by hengGuo
    private static ConcurrentHashMap<String, Socket> nodehostIp_sockets = new ConcurrentHashMap<>();
    private static int DataSrcCount = 0;
    private static int DataRevCount = 0;
    private static final String LOCAL_IP = "127.0.0.1";
    private static Socket socketCmd8099 = null;   // 用于向dataAccess发送命令包(CmdFrame)，启动新任务或结束旧任务。

    @Value("${simserver.baseurl}")
    private String serverBaseUrl;

    @Value("${node.id}")
    private int nodeId;

    @Value("${node.ip}")
    private String nodeIp;

    @Value("${datasource.ip}")
    private String dataSourceIp;
    @Value("${datasource.port}")
    private int dataSourcePort;

    @Value("${context.schema-run-dir}")
    private String schemaRunDir;

    @Value("${context.components-dir}")
    private String componentsDir;

    @Value("${context.common-include-dir}")
    private String commonIncludeDir;

    @Value("${context.common-src-dir}")
    private String commonSrcDir;

    @Value("${context.common-bin-dir}")
    private String commonBinDir;

    @Value("${context.java-dir}")
    private String javaDir;

    @Value("${dataAccess.ip}")
    private String dataAccessIp;
    @Value("${dataAccess.port}")
    private int dataAccessPort;

    @Autowired
    private Dispatcher dispatcher;

    @Async
    public void runListener(Listener listener) {
        try (ServerSocket serverSocket = new ServerSocket(listener.getPort())) {
            while (true) {
                Socket clientSocket = serverSocket.accept();
                clientSocket.setSoTimeout(listener.getTimeout() * 1000);
                InetSocketAddress clientAddress = (InetSocketAddress) clientSocket.getRemoteSocketAddress();
                System.out.println(clientAddress.toString() + " connected");
//                AsyncTaskService ts = NodeHostApplication.ac.getBean(AsyncTaskService.class);
//                ts.runReceiver(clientSocket, listener.getMaxFrameLength(), listener.getDispatcher());
//                ts.runReceiver_bak(clientSocket, listener.getDispatcher());
                new Thread(() -> {
                    try {
                        runReceiver_bak_gh(clientSocket, dispatcher);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }).start();
                System.out.println("receive thread running...");
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

/*    @Async
    public void runReceiver(Socket clientSocket, int maxFrameLength, Dispatcher dispatcher) {
        try {
            //OutputStream os=clientSocket.getOutputStream();
            InputStream is = clientSocket.getInputStream();
            Parser parser = new Parser(maxFrameLength, dispatcher, schemaRunDir);
            byte[] buf = new byte[BUF_LENGTH];

            //receive data frame
            int n = 0;
            while (n >= 0) {
                try {
                    n = is.read(buf);
                    if (n > 0) {
                        //dispatch data frame to schema
                        //System.out.format("recv %d bytes\n",n);
                        if (parser.parse(buf, n) < 0)
                            throw new Exception("parse data error.");
                    }
                } catch (SocketTimeoutException ste) {
                    System.out.println("read timeout");
                } catch (SocketException se) {
                    System.out.println("scoket exception ");
                    n = -1;
                }
            }
            System.out.println("client disconnected");
            clientSocket.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }*/

    @Async
    public void runReceiver_bak(Socket clientSocket, Dispatcher dispatcher) throws IOException {
        System.out.println("getin method BAK");
        int count_FrameNumber = 1;
        while (true) {

            DataFrame receivedFrame = DataFrame.readOneFrame(clientSocket); //  收到一帧数据
            log.info(receivedFrame.toString());
            SchemaExecutor schemaExecutor = dispatcher.getSchemaExecutor(receivedFrame.getSchemaId()); // 依据帧号获取模式执行器schemaExecutor
            ArrayList<JSONObject> links = schemaExecutor.getSchema().getLinks();    //  通过模式执行器获取Link信息
            System.out.println(links);
            String result_UUID = schemaExecutor.getSchema().getResult_uuid();
            log.info("Result_UUID: " + result_UUID);
//            if(receivedFrame.getSourceComponentId())

            //  接受结果组件，将结果组件数据进行本地化存储,并计数
            if (receivedFrame.getSourceComponentId().equals(result_UUID)) {
                DataRevCount++;
                String result = new String(receivedFrame.getData());
                File folder = new File(String.valueOf(receivedFrame.getSchemaId()));
                if (!folder.exists()) {
                    folder.mkdir();
                }
                String fileName = receivedFrame.getSchemaId() + "/" + count_FrameNumber + ".txt";
                File file = new File(fileName);
                try {
                    FileWriter fileWriter = new FileWriter(file);
                    fileWriter.write(result);
                    fileWriter.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
                count_FrameNumber++;
//                if(DataRevCount %2 ==0){
//                    HttpRequestUtils.ProgressPost(serverBaseUrl, DataRevCount);
//                }


            }

            //  判断sockets映射集合是否存入源组件链接，未存入则存入
            if (!componentId_sockets.containsKey(receivedFrame.getSourceComponentId())) {
                componentId_sockets.put(receivedFrame.getSourceComponentId(), clientSocket);
                log.info("componentId_socketName: " + receivedFrame.getSourceComponentId() + " componenetId_size+: " + componentId_sockets.size());
            }


            OutputStream outputStream = null;    //  初始化输出流
            for (String id : componentId_sockets.keySet()) {    //遍历sockets映射集合
                if (receivedFrame.getSourceComponentId().equals(id)) {   //确认收到帧的源组件ID在sockets集合内
                    System.out.println(links);
                    for (JSONObject link : links) { //  遍历Link集合寻找下一个对应关系
                        log.info("PORT1: " + link.getJSONObject("source").getString("port")
                                + "PORT2 " + receivedFrame.getSourceComponentPort());

                        //  匹配link的port名和收到帧的源组件端口名是否一致
                        if (receivedFrame.getSourceComponentPort().equals(link.getJSONObject("source").getString("port"))) {
                            //  一致时修改目的组件名和谜底端口号
                            String TargetPort = link.getJSONObject("target").getString("port");
                            String TargetComId = link.getJSONObject("target").getString("id");
                            receivedFrame.setTargetComponentPort(TargetPort);
                            receivedFrame.setTargetComponentId(TargetComId);
                            outputStream = componentId_sockets.get(TargetComId).getOutputStream();
                            log.info(outputStream.toString());
                            System.out.println(receivedFrame);
                        }
                    }
                }

                if (outputStream != null) {
                    byte[] serializedFrame = receivedFrame.serialize();
                    outputStream.write(serializedFrame);
                    outputStream.flush();
                }
            }
        }

    }

    // Add by hengGuo
    @Async
    public void runReceiver_bak_gh(Socket clientSocket, Dispatcher dispatcher) throws IOException {
        System.out.println("getin method BAK");
        int count_FrameNumber = 1;
        while (true) {

            DataFrame receivedFrame = DataFrame.readOneFrame(clientSocket); //  收到一帧数据
//          todo-if(receivedFrame.getSourceComponentId().isEmpty()) continue;
            log.info("接收时立刻找帧：" + receivedFrame.toString());
            SchemaExecutor schemaExecutor = dispatcher.getSchemaExecutor(receivedFrame.getSchemaId()); // 依据帧号获取模式执行器schemaExecutor
            ArrayList<JSONObject> links = schemaExecutor.getSchema().getLinks();    //  通过模式执行器获取Link信息
//            System.out.println(links); 空指针异常
            String result_UUID = schemaExecutor.getSchema().getResult_uuid();
            log.info("Result_UUID: " + result_UUID);
            // to test1
            if (receivedFrame == null) {
                log.error("reveivedFrame is null");
            }
            if (receivedFrame.getSourceComponentId() == null) {
                log.error("reveivedFrame getSourceComponentId is null");
            }

            //  接受结果组件，将结果组件数据进行本地化存储,并计数
            if (receivedFrame.getSourceComponentId().equals(result_UUID)) {
                DataRevCount++;
                String result = new String(receivedFrame.getData());
                File folder = new File(String.valueOf(receivedFrame.getSchemaId()));
                if (!folder.exists()) {
                    folder.mkdir();
                }
                String fileName = receivedFrame.getSchemaId() + "/" + count_FrameNumber + ".txt";
                File file = new File(fileName);
                try {
                    FileWriter fileWriter = new FileWriter(file);
                    fileWriter.write(result);
                    fileWriter.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
                count_FrameNumber++;
//                if(DataRevCount %2 ==0){
//                    HttpRequestUtils.ProgressPost(serverBaseUrl, DataRevCount);
//                }
            }

            //  判断sockets映射集合是否存入源组件链接，未存入则存入
            if (!componentId_sockets.containsKey(receivedFrame.getSourceComponentId())) {
                componentId_sockets.put(receivedFrame.getSourceComponentId(), clientSocket);
                log.info("componentId_socketName: " + receivedFrame.getSourceComponentId() + " componenetId_size+: " + componentId_sockets.size());
            }


            OutputStream outputStream = null;    //  初始化输出流

//            String componentId = receivedFrame.getSourceComponentId();
//            if (componentId_sockets.containsKey(componentId)) {    //  判断sockets映射集合是否存入源组件链接，已存入则获取输出流
//                outputStream = componentId_sockets.get(componentId).getOutputStream();
//            }
//            else
//            {
//                System.out.println("component socket don't be built");
//            }

//            System.out.println(links);
            for (JSONObject link : links) { //  遍历Link集合寻找下一个对应关系
                log.info("PORT1: " + link.getJSONObject("source").getString("port")
                        + "PORT2 " + receivedFrame.getSourceComponentPort());

                //  匹配link的port名和收到帧的源组件端口名是否一致
                if (receivedFrame.getSourceComponentPort().equals(link.getJSONObject("source").getString("port"))) {
                    //  一致时修改目的组件名和谜底端口号
                    String TargetPort = link.getJSONObject("target").getString("port");
                    String TargetComId = link.getJSONObject("target").getString("id");
                    receivedFrame.setTargetComponentPort(TargetPort);
                    receivedFrame.setTargetComponentId(TargetComId);
                    log.info("wrong before : " + receivedFrame);
                    // to test
                    if (componentId_sockets.get(TargetComId) == null) continue;
                    Socket tmpsocket = componentId_sockets.get(TargetComId);
                    outputStream = tmpsocket.getOutputStream();
//                    log.info(outputStream.toString());
//                    System.out.println(receivedFrame);
                    break;
                }
            }
            // componentId_socket存储组件id和组件socket的关系
            // componentnodemap存储组件id和nodehostIP的关系
            // nodehostIp_sockets存储nodehostIP和nodehostsocket的关系
            // 填充好targetComponentId后
            // 根据组件id确定nodeip，从而确定是否转发到其他nodehost，如果不进行跨nodehost的转发，则需要转发到其他组件，
            // 跨nodehost转发需要用到nodehostIP，
            // 不跨nodehost进行转发只需要将包转发给组件的socket就行了
            System.out.println("receivedFrame.getTargetComponentId() = " + receivedFrame.getTargetComponentId());
            //获得目标组件的IP
            System.out.println("schemaExecutor.getSchema().getComponentnodemap() = " + schemaExecutor.getSchema().getComponentnodemap());
            String nodeip = schemaExecutor.getSchema().getComponentnodemap().get(receivedFrame.getTargetComponentId());

            System.out.println("nodeip="+nodeip+"---");//todo:nodeip有问题,是ID

            if (nodeip == null) {
                //因为有空包的存在，所以只要nodeip为空就意味着不需要发任何东西，直接读取下一个包
                continue;
            }
            if (nodeip.equals(nodeIp)) {
                // 如果是本机组件，什么也不需要做，最后统一发
                System.out.println("本机组件转发");
            } else {
                // 根据nodeip创建socket链接
                Socket socket = null;
                if (nodehostIp_sockets.containsKey(nodeip)) {
                    //使用componectId_sockets中的socket向目标nodeip发送数据 转发数据到nodeHost时不需要进行解帧
                    socket = nodehostIp_sockets.get(nodeip);
                } else {
                    //向nodeip发送socket请求，并加入hash表
                    socket = new Socket(nodeip, 9999);
                    nodehostIp_sockets.put(nodeip, socket);
                }
                System.out.println("跨nodehost进行转发");
                outputStream = socket.getOutputStream();
            }
            // 转发(根据outputStream的不同选择转发给其他nodehost还是转发给其他组件)
            if (outputStream != null) {
                byte[] serializedFrame = receivedFrame.serialize();
                outputStream.write(serializedFrame);
                outputStream.flush();
            } else {
                System.out.println("node ip is not null but outputStream is null");
            }
        }
    }

    // Add by hengGuo
    // Suggest to check ip but not achieved
    @Async
    public void ListenToOtherNodeHost(Socket clientSocket, Dispatcher dispatcher) throws IOException {

        System.out.println("收到其他nodehost转发请求，getin method BAK");
        int count_FrameNumber = 1;
        while (true) {

            DataFrame receivedFrame = DataFrame.readOneFrame(clientSocket); //  收到一帧数据
            log.info("接收时立刻找帧：" + receivedFrame.toString());
            SchemaExecutor schemaExecutor = dispatcher.getSchemaExecutor(receivedFrame.getSchemaId()); // 依据帧号获取模式执行器schemaExecutor
            ArrayList<JSONObject> links = schemaExecutor.getSchema().getLinks();    //  通过模式执行器获取Link信息
            System.out.println(links);
            String result_UUID = schemaExecutor.getSchema().getResult_uuid();
            log.info("Result_UUID: " + result_UUID);
            // to test1
            if (receivedFrame == null) {
                log.error("reveivedFrame is null");
            }
            if (receivedFrame.getSourceComponentId() == null) {
                log.error("reveivedFrame getSourceComponentId is null");
            }

            //  接受结果组件，将结果组件数据进行本地化存储,并计数
            if (receivedFrame.getSourceComponentId().equals(result_UUID)) {
                DataRevCount++;
                String result = new String(receivedFrame.getData());
                File folder = new File(String.valueOf(receivedFrame.getSchemaId()));
                if (!folder.exists()) {
                    folder.mkdir();
                }
                String fileName = receivedFrame.getSchemaId() + "/" + count_FrameNumber + ".txt";
                File file = new File(fileName);
                try {
                    FileWriter fileWriter = new FileWriter(file);
                    fileWriter.write(result);
                    fileWriter.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
                count_FrameNumber++;
//                if(DataRevCount %2 ==0){
//                    HttpRequestUtils.ProgressPost(serverBaseUrl, DataRevCount);
//                }
            }

            //  判断sockets映射集合是否存入源组件链接，未存入则存入
            if (!componentId_sockets.containsKey(receivedFrame.getSourceComponentId())) {
                componentId_sockets.put(receivedFrame.getSourceComponentId(), clientSocket);
                log.info("componentId_socketName: " + receivedFrame.getSourceComponentId() + " componenetId_size+: " + componentId_sockets.size());
            }


            OutputStream outputStream = null;    //  初始化输出流

//            String componentId = receivedFrame.getSourceComponentId();
//            if (componentId_sockets.containsKey(componentId)) {    //  判断sockets映射集合是否存入源组件链接，已存入则获取输出流
//                outputStream = componentId_sockets.get(componentId).getOutputStream();
//            }
//            else
//            {
//                System.out.println("component socket don't be built");
//            }

            System.out.println(links);
            for (JSONObject link : links) { //  遍历Link集合寻找下一个对应关系
                log.info("PORT1: " + link.getJSONObject("source").getString("port")
                        + "PORT2 " + receivedFrame.getSourceComponentPort());

                //  匹配link的port名和收到帧的源组件端口名是否一致
                if (receivedFrame.getSourceComponentPort().equals(link.getJSONObject("source").getString("port"))) {
                    //  一致时修改目的组件名和谜底端口号
                    String TargetPort = link.getJSONObject("target").getString("port");
                    String TargetComId = link.getJSONObject("target").getString("id");
                    receivedFrame.setTargetComponentPort(TargetPort);
                    receivedFrame.setTargetComponentId(TargetComId);
                    log.info("wrong before : " + receivedFrame);
                    // to test
                    if (componentId_sockets.get(TargetComId) == null) continue;
                    Socket tmpsocket = componentId_sockets.get(TargetComId);
                    outputStream = tmpsocket.getOutputStream();
                    log.info(outputStream.toString());
                    System.out.println(receivedFrame);
                    break;
                }
            }
            // componentId_socket存储组件id和组件socket的关系
            // componentnodemap存储组件id和nodehostIP的关系
            // nodehostIp_sockets存储nodehostIP和nodehostsocket的关系
            // 填充好targetComponentId后
            // 根据组件id确定nodeip，从而确定是否转发到其他nodehost，如果不进行跨nodehost的转发，则需要转发到其他组件，
            // 跨nodehost转发需要用到nodehostIP，
            // 不跨nodehost进行转发只需要将包转发给组件的socket就行了
            System.out.println("receivedFrame.getTargetComponentId() = " + receivedFrame.getTargetComponentId());
            //获得目标组件的IP
            String nodeip = schemaExecutor.getSchema().getComponentnodemap().get(receivedFrame.getTargetComponentId());
            if (nodeip == null) {
                //因为有空包的存在，所以只要nodeip为空就意味着不需要发任何东西，直接读取下一个包
                continue;
            }
            if (nodeip.equals(nodeIp)) {
                // 如果是本机组件，什么也不需要做，最后统一发
                System.out.println("本机组件转发");
            } else {
                // 根据nodeip创建socket链接
                Socket socket = null;
                if (nodehostIp_sockets.containsKey(nodeip)) {
                    //使用componectId_sockets中的socket向目标nodeip发送数据 转发数据到nodeHost时不需要进行解帧
                    socket = nodehostIp_sockets.get(nodeip);
                } else {
                    //向nodeip发送socket请求，并加入hash表
                    socket = new Socket(nodeip, 9999);
                    nodehostIp_sockets.put(nodeip, socket);
                }
                System.err.println("跨nodehost进行转发，这里是不可能的！！！");
                outputStream = socket.getOutputStream();
            }
            // 转发(根据outputStream的不同选择转发给其他nodehost还是转发给其他组件)
            if (outputStream != null) {
                byte[] serializedFrame = receivedFrame.serialize();
                outputStream.write(serializedFrame);
                outputStream.flush();
            } else {
                System.out.println("node ip is not null but outputStream is null");
            }
        }
    }


    public boolean sendDataControlPacket(JSONObject dataControlPacket) {
        boolean result = true;
        try {
            InetAddress addr = InetAddress.getByName(dataSourceIp);
            DatagramSocket ds = new DatagramSocket();

            byte[] packetData = new byte[1024];//maybe not enough space
            int packetLen = 0;
            packetData[packetLen++] = 0x11;
            Utils.int2bytes(dataControlPacket.getInt("schemaId"), packetData, packetLen);
            packetLen += 4;
            Utils.int2bytes(dataControlPacket.getInt("sendNumber"), packetData, packetLen);
            packetLen += 4;
            Utils.int2bytes(dataControlPacket.getInt("portNumber"), packetData, packetLen);
            packetLen += 4;
            packetLen += 4;//reserved
            for (int i = 0; i < dataControlPacket.getInt("portNumber"); i++) {
                JSONObject component = dataControlPacket.getJSONArray("components").getJSONObject(i);
                byte[] uuid = component.getString("uuid").getBytes();
                System.arraycopy(uuid, 0, packetData, packetLen, uuid.length < 36 ? uuid.length : 36);
                packetLen += 36;
                byte[] port = component.getString("port").getBytes();
                System.arraycopy(port, 0, packetData, packetLen, port.length < 64 ? port.length : 64);
                packetLen += 64;
                Utils.int2bytes(component.getInt("rows"), packetData, packetLen);
                packetLen += 4;
                Utils.int2bytes(component.getInt("cols"), packetData, packetLen);
                packetLen += 4;
                Utils.int2bytes(component.getInt("type"), packetData, packetLen);
                packetLen += 4;
                packetLen += 8;//reserved
            }
            packetData[packetLen++] = (byte) 0xFF;
            DatagramPacket dp = new DatagramPacket(packetData, packetLen, addr, dataSourcePort);
            ds.send(dp);
            ds.close();
            packetData = null;
        } catch (Exception e) {
            e.printStackTrace();
            result = false;
        }
        return result;
    }

    @Async
    public void runSchema(JSONObject schemaJson) {

        try {
//            System.out.println("----shemaJson:" + schemaJson.toString());
            Schema schema = (Schema) JSONObject.toBean(schemaJson, Schema.class);  //将JSON对象转换为Schema类的实例
            SchemaExecutor executor = new SchemaExecutor(schema, dispatcher);
            JSONObject executeContext = new JSONObject();

            System.out.println("schema=" + schema);

            executeContext.put("nodeIp", nodeIp);
            executeContext.put("serverBaseUrl", serverBaseUrl);
            executeContext.put("schemaRunDir", schemaRunDir);
            executeContext.put("componentsDir", componentsDir);
            executeContext.put("commonIncludeDir", commonIncludeDir);
            executeContext.put("commonSrcDir", commonSrcDir);
            executeContext.put("commonBinDir", commonBinDir);

            if (!executor.init(executeContext))        //status=1
                throw new Exception("schema " + schema.getId() + "init fail");
//            if (!executor.execute()) {       //status=3
//                throw new Exception("schema " + schema.getId() + " start execute fail");
//            }

            if (!executor.execute_docker()) {       //status=3
                throw new Exception("schema " + schema.getId() + " start execute fail");
            }
//            executor.close();       //status=0  接收到停止指令和超时时关闭执行器
//            System.out.println("schema " + schema.getId() + " is finished");

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Async
    public void runComponent(ComponentExecutor executor) {
        try {
            String command4j = "java " + javaDir + "\\" + executor.getId();


            String program = executor.getProgram();
            Vector<String> command = new Vector<>();//cmdlineToCommand(compileCmdLine,compileDir);
            command.add(program);
            //command.add(""+executor.getNextFrameNo());
            for (String arg : executor.getCommandArgs()) {
                command.add(arg);
            }
            File programOutputFile = new File(executor.getProgramDir() + "/" + executor.getComponentId() + ".err");
            ProcessBuilder pb = new ProcessBuilder(command);
            Map<String, String> env = pb.environment();
            pb.redirectError(programOutputFile);
            pb.directory(new File(executor.getProgramDir()));
            //pb.redirectErrorStream(true);
            Process p = pb.start();
            BufferedReader input = new BufferedReader(new InputStreamReader(p.getInputStream()));
            PrintStream out = new PrintStream(p.getOutputStream());
            executor.setStatus(SchemaExecutor.RUNNING);
            while (!executor.isNeedTerminate() && !executor.finished()) {
				/*
				if(executor.getComponentName().equals("SVG"))
					System.out.println("debug svg");
				*/
                if (executor.isNeedPause()) {
                    if (executor.getStatus() != SchemaExecutor.PAUSED) {
                        executor.setStatus(SchemaExecutor.PAUSED);
                        System.out.println("component " + executor.getComponentName() + " [" + executor.getComponentId() + "] paused");
                    }
                    Thread.sleep(100);
                    continue;
                }
                executor.setStatus(SchemaExecutor.RUNNING);
                if (executor.canRunOnce()) {
                    //switchToRoot();
                    int frameNo = executor.getNextFrameNo();
                    System.out.println("component " + executor.getComponentName() + " [" + executor.getComponentId() + "] running for frame " + frameNo);
                    out.println(frameNo);
                    out.flush();
                    long startTime = System.currentTimeMillis();
                    String line = input.readLine();
                    System.out.println(line);
                    long finishTime = System.currentTimeMillis();
                    System.out.println("component " + executor.getComponentName() + "[" + executor.getComponentId() + "] running for frame " + executor.getNextFrameNo() + " finished, consumed " + (finishTime - startTime) / 1000.0 + " seconds");
                    executor.finishRun();
                } else {
                    //System.out.println("component "+executor.getComponentId()+" is waiting data...");
                    Thread.sleep(100);
                }
            }
            //send 0 to terminate program
            out.println(0);
            out.flush();
            String line = input.readLine();
            System.out.println(line);
            input.close();
            out.close();
            //wait program terminated
            boolean terminated = false;
            int exitValue = 0;
            int waitTime = 0;
            while (!terminated) {//&&waitTime<MAX_COMPILE_TIME*1000
                try {
                    exitValue = p.exitValue();

                    terminated = true;
                } catch (IllegalThreadStateException e) {
                    waitTime += 100;
                    Thread.sleep(100);
                }

            }
            System.out.println("component process teminated, exitValue=" + exitValue);
            executor.setStatus(SchemaExecutor.IDLE);
            System.out.println("component executor " + executor.getComponentName() + "[" + executor.getComponentId() + "] is finished");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Value("${harbor.ip}")
    String harborIp;
    @Value("${harbor.project-name}")
    String testProject;
    @Value("${harbor.username}")
    String username;
    @Value("${harbor.password}")
    String password;
    @Async
    //添加runComponentDokcer
    public void runComponentDocker(ComponentExecutor executor, String dockerImage, String uniqueDockerName) {
        try {
            //不能加-d,不要隐藏log输出
            String flag ="docker login "+harborIp+" -u "+username+" -p "+password+" && "+
                    "docker pull "+harborIp+"/"+testProject+"/"+dockerImage.toLowerCase()+":latest"+" && "+
                    "docker run -t --net=host --name " + uniqueDockerName +" " +harborIp+"/"+testProject+"/"+dockerImage.toLowerCase()+":latest"+
                    " /bin/java " + dockerImage + " " + executor.getSchemaExecutor().getSchemaId() + " "
                    + uniqueDockerName + " 9999";
            log.info("##################docker run command:"+flag);
            String[] dockerCMD = {"sh", "-c", flag};
            Process dockerRun = Runtime.getRuntime().exec(dockerCMD);
            log.info("##################start docker success!");

            System.out.println("###########uniqueDockerName"+uniqueDockerName);
            executor.setComponentDocker(uniqueDockerName);    //为每个组件执行器绑定进程

            new Thread(() ->{
                File programOutputErrorFile = new File(javaDir + dockerImage + "/" + dockerImage + ".err");
                File programOutputNormalFile = new File(javaDir + dockerImage + "/" + dockerImage + ".txt");

                if (programOutputNormalFile.getParentFile() != null) {
                    // 如果父目录不存在，创建所有必要的目录
                    programOutputNormalFile.getParentFile().mkdirs();
                }

                try {
                    // 创建文件
                    if (programOutputNormalFile.createNewFile()) {
                        System.out.println("文件创建成功: " + programOutputNormalFile.getAbsolutePath());
                    } else {
                        System.out.println("文件已存在: " + programOutputNormalFile.getAbsolutePath());
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }

                if (programOutputErrorFile.getParentFile() != null) {
                    // 如果父目录不存在，创建所有必要的目录
                    programOutputErrorFile.getParentFile().mkdirs();
                }

                try {
                    // 创建文件
                    if (programOutputErrorFile.createNewFile()) {
                        System.out.println("文件创建成功: " + programOutputErrorFile.getAbsolutePath());
                    } else {
                        System.out.println("文件已存在: " + programOutputErrorFile.getAbsolutePath());
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }


                //读取正常输出  当dockerRun正常执行时，容器内java程序的错误会返回为dockerRun的programOutputNormalFile
                BufferedReader reader = new BufferedReader(new InputStreamReader(dockerRun.getInputStream()));
                try (FileWriter normalwriter = new FileWriter(programOutputNormalFile)) {
                    String line1;
                    while ((line1 = reader.readLine()) != null) {
//                    System.out.println(line1);
                        normalwriter.write(line1 + System.lineSeparator()); // 写入文件并换行
                    }
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
                // 读取错误输出
                BufferedReader errorReader = new BufferedReader(new InputStreamReader(dockerRun.getErrorStream()));
                try (FileWriter errwriter = new FileWriter(programOutputErrorFile)) {
                    String line;
                    while ((line = errorReader.readLine()) != null) {
//                    System.out.println(line);
                        errwriter.write(line + System.lineSeparator()); // 写入文件并换行
                    }
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            }) .start() ;
//            executor.setStatus(SchemaExecutor.RUNNING);
            //没用
//            while (!executor.finished()) {
//                executor.setStatus(SchemaExecutor.RUNNING);
//            }

//            int exitValue = 0;
//            try {
//                exitValue = dockerRun.exitValue();
//            } catch (IllegalThreadStateException e) {
//                System.out.println("component process has not yet teminated, exitValue=" + exitValue);
//            }
//            executor.setStatus(SchemaExecutor.IDLE);
//            System.out.println("component executor " + executor.getComponentName() + "[" + executor.getComponentId() + "] is finished");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    //添加runComponentJava
    public void runComponent_bak(ComponentExecutor executor, String ComponentID, String ComponentUUID) {
        try {
            log.info(ComponentID);
            String command4j = javaDir + "\\" + executor.getId();
            ArrayList<String> command = new ArrayList<>();//cmdlineToCommand(compileCmdLine,compileDir);
            File programOutputErrorFile = new File(javaDir + ComponentID + ".err");
            File programOutputNormalFile = new File(javaDir + ComponentID + ".txt");


            ProcessBuilder pb = new ProcessBuilder("java", ComponentID, String.valueOf(executor.getSchemaExecutor().getSchemaId()), ComponentUUID, "9999");
            pb.directory(new File(javaDir));
            Map<String, String> env = pb.environment();
            System.out.println(env);
            pb.redirectError(programOutputErrorFile);
            pb.redirectOutput(programOutputNormalFile);
            //  pb.directory(new File(javaDir));
            //  pb.redirectErrorStream(true);
            Process p = pb.start();
            executor.setComponentProcess(p);    //为每个组件执行器绑定进程
            log.info(String.valueOf(p));


            BufferedReader input = new BufferedReader(new InputStreamReader(p.getInputStream()));   //返回P进程的输出
            PrintStream out = new PrintStream(p.getOutputStream());
            executor.setStatus(SchemaExecutor.RUNNING);
            //等待组件执行器执行完毕或者提前结束
            //todo-考虑isneedTerminate要不要删掉
            while (!executor.finished()) {
                executor.setStatus(SchemaExecutor.RUNNING);
            }


//            while (!executor.isNeedTerminate() && !executor.finished()) {
//				/*
//				if(executor.getComponentName().equals("SVG"))
//					System.out.println("debug svg");
//				*/
//                executor.setStatus(SchemaExecutor.RUNNING);
//            }

            //send 0 to terminate program
            out.println(0);
            out.flush();
            String line = input.readLine();
            System.out.println(line);
            input.close();
            out.close();
            //wait program terminated
//            boolean terminated = false;
//            int exitValue = 0;
//            int waitTime = 0;
//            while (!terminated) { //&&waitTime<MAX_COMPILE_TIME*1000
//                try {
//                    exitValue = p.exitValue();
//
//                    terminated = true;
//                } catch (IllegalThreadStateException e) {
//                    waitTime += 100;
//                    Thread.sleep(100);
//                }
//
//            }
            int exitValue = 0;
            try {
                exitValue = p.exitValue();
            } catch (IllegalThreadStateException e) {
                System.out.println("component process has not yet teminated, exitValue=" + exitValue);
            }
            executor.setStatus(SchemaExecutor.IDLE);
            System.out.println("component executor " + executor.getComponentName() + "[" + executor.getComponentId() + "] is finished");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Async
    // 这是一个异步函数，用于连接到一个指定的IP地址和端口。如果连接失败，将会打印错误信息并继续尝试连接，直到成功连接为止
    public void Connect2DataAccess() {
        boolean connected = false;
        while (!connected) {
            try {
                socketCmd8099 = new Socket(dataAccessIp, dataAccessPort);
                connected = true;
                log.info("nodeHost is try to connect dataAccess By ): " + socketCmd8099);
            } catch (IOException e) {

                log.info("nodeHost is not connected to dataAccess,  socketCmd8099): " + socketCmd8099);

                e.printStackTrace();
            }
        }
        log.info(" nodeHost is connected to dataAccess,  socketCmd8099): " + socketCmd8099);
    }

    @Async
    //  发送命令帧到socketCmd8099
    public void sendCmdFrame2DataAccess(int SchemaId, int DataRemote_Port, int SrcFrameLength, int StartStopFlag) {
        CmdFrame cf = new CmdFrame();
        cf.setSchemaId(SchemaId);
        cf.setSrcFrameLength(SrcFrameLength);
        cf.setPort_dataRemote(DataRemote_Port);  // dataSrc will send data via 8888.
        cf.setDataAccess_StartStop(StartStopFlag);  // start_stop


        // 发送给dataAccess.
        try {
            OutputStream out = socketCmd8099.getOutputStream();
            out.write(cf.serialize());
            out.flush();
            System.out.println("##############sendCmdFrame2DataAccess Success!");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


}
